// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License.  You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use anyhow::{anyhow, Result};
use skywalking::reporter::{CollectItemConsume, CollectItemProduce, grpc::GrpcReporter};
use std::{
    sync::atomic::{AtomicU32, Ordering},
    time::{Duration, SystemTime},
};
use tokio::time::sleep;
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity};
use tracing::{debug, error, info, warn, trace};

// 最小重试间隔，初始为1秒
const MIN_RETRY_INTERVAL_MS: u64 = 1000;
// 最大重试间隔，5分钟
const MAX_RETRY_INTERVAL_MS: u64 = 300000;
// 重试计数衰减周期，30秒
const RETRY_COUNT_DECAY_PERIOD_MS: u64 = 30000;
// 重试成功后的稳定期，3分钟内不会增加重试间隔
const STABLE_PERIOD_MS: u64 = 180000;

// 重试状态跟踪
static RETRY_COUNT: AtomicU32 = AtomicU32::new(0);
static LAST_SUCCESS: AtomicU32 = AtomicU32::new(0);

pub struct GrpcReporterConfiguration {
    pub server_addr: String,
    pub authentication: String,
    pub enable_tls: bool,
    pub ssl_cert_chain_path: String,
    pub ssl_key_path: String,
    pub ssl_trusted_ca_path: String,
}

pub async fn run_reporter(
    config: GrpcReporterConfiguration, producer: impl CollectItemProduce,
    consumer: impl CollectItemConsume,
) -> Result<()> {
    info!("Starting gRPC reporter with server address: {}", config.server_addr);
    debug!("gRPC configuration: auth_enabled={}, tls_enabled={}", 
           !config.authentication.is_empty(), config.enable_tls);
    
    let endpoint = create_endpoint(&config).await?;
    trace!("Created endpoint: {:?}", endpoint);
    
    info!("Connecting to OAP server...");
    let channel = connect(endpoint).await;
    debug!("Connected to OAP server, creating reporter");

    let mut reporter = GrpcReporter::new_with_pc(channel, producer, consumer);
    trace!("Created base gRPC reporter");

    if !config.authentication.is_empty() {
        debug!("Configuring authentication for reporter");
        reporter = reporter.with_authentication(config.authentication.clone());
    }

    // 设置更大的gRPC通道容量和并发处理
    debug!("Configuring reporter parameters");
    reporter = reporter
        .with_batch_size(1000)                           // 增加批量处理大小
        .with_flush_interval(Duration::from_secs(3))     // 减少刷新间隔，更频繁地发送数据
        .with_base_uri("http://skywalking-oap-server")   // 设置基础URI，用于日志
        .with_timeout(Duration::from_secs(60));          // 设置请求超时
    
    // 增加收集超时设置
    debug!("Configuring collection timeouts (60s each)");
    reporter = reporter
        .with_trace_collect_timeout(Duration::from_secs(60))
        .with_log_collect_timeout(Duration::from_secs(60))
        .with_meter_collect_timeout(Duration::from_secs(60));

    info!("Reporter configuration complete, worker is ready");

    debug!("Setting up status handler for error monitoring");
    let handle = reporter
        .reporting()
        .await
        .with_status_handle(|message, status| {
            let code = status.code();
            
            // 详细记录错误状态信息
            trace!("Received gRPC status: code={:?}, message={}, details={:?}", 
                  code, status.message(), status.details());
            
            // 检查错误是否是连接问题
            let is_connection_error = matches!(
                code,
                tonic::Code::Unavailable | 
                tonic::Code::DeadlineExceeded | 
                tonic::Code::Aborted
            );
            
            if is_connection_error {
                // 增加重试计数
                let prev_count = RETRY_COUNT.fetch_add(1, Ordering::SeqCst);
                error!(
                    ?status, 
                    retry_count = prev_count + 1,
                    "Connection error to OAP server: {} (code: {:?})", 
                    message, code
                );
                
                // 添加更详细的错误源信息
                if let Some(source) = status.source() {
                    error!("Error source details: {:?}", source);
                }
            } else {
                warn!(
                    ?status, 
                    "Collect failed: {} (code: {:?})", 
                    message, code
                );
                
                // 添加更详细的错误源信息
                if let Some(source) = status.source() {
                    debug!("Error source details: {:?}", source);
                }
            }
        })
        .spawn();

    // 启动状态监控任务
    debug!("Spawning connection health monitor");
    tokio::spawn(connection_health_monitor());

    info!("Reporter is running, waiting for completion");
    handle
        .await
        .map_err(|err| anyhow!("Tracer reporting failed: {:?}", err))?;

    Ok(())
}

async fn create_endpoint(config: &GrpcReporterConfiguration) -> Result<Endpoint> {
    let scheme = if config.enable_tls { "https" } else { "http" };

    let url = format!("{}://{}", scheme, config.server_addr);
    debug!(url, "Creating gRPC endpoint");
    let mut endpoint = Endpoint::from_shared(url)?;
    
    // 为gRPC设置超时和连接参数，提高可靠性
    trace!("Configuring gRPC endpoint connection parameters");
    endpoint = endpoint
        .timeout(Duration::from_secs(10))
        .connect_timeout(Duration::from_secs(5))
        .tcp_keepalive(Some(Duration::from_secs(30)))
        .tcp_nodelay(true)
        .http2_keep_alive_interval(Duration::from_secs(30))
        .keep_alive_timeout(Duration::from_secs(15))
        .keep_alive_while_idle(true);

    debug!(
        enable_tls = config.enable_tls,
        ssl_trusted_ca_path = config.ssl_trusted_ca_path,
        ssl_key_path = config.ssl_key_path,
        ssl_cert_chain_path = config.ssl_cert_chain_path,
        "Configuring TLS settings"
    );

    if config.enable_tls {
        let domain_name = config.server_addr.split(':').next().unwrap_or_default();
        debug!(domain_name, "Configuring TLS domain");
        let mut tls = ClientTlsConfig::new().domain_name(domain_name);

        let ssl_trusted_ca_path = &config.ssl_trusted_ca_path;
        if !ssl_trusted_ca_path.is_empty() {
            debug!(ssl_trusted_ca_path, "Configuring TLS CA certificate");
            let ca_cert = tokio::fs::read(&config.ssl_trusted_ca_path).await?;
            trace!("Read CA certificate, size: {} bytes", ca_cert.len());
            let ca_cert = Certificate::from_pem(ca_cert);
            tls = tls.ca_certificate(ca_cert);
        }

        let ssl_key_path = &config.ssl_key_path;
        let ssl_cert_chain_path = &config.ssl_cert_chain_path;
        if !ssl_key_path.is_empty() && !ssl_cert_chain_path.is_empty() {
            debug!("Configuring mTLS with client certificate and key");
            let client_cert = tokio::fs::read(&config.ssl_cert_chain_path).await?;
            trace!("Read client certificate, size: {} bytes", client_cert.len());
            let client_key = tokio::fs::read(&config.ssl_key_path).await?;
            trace!("Read client key, size: {} bytes", client_key.len());
            let client_identity = Identity::from_pem(client_cert, client_key);
            tls = tls.identity(client_identity);
        }

        endpoint = endpoint.tls_config(tls)?;
        debug!("TLS configuration complete");
    }

    debug!("Endpoint creation complete");
    Ok(endpoint)
}

// 计算当前的重试等待时间（使用指数退避算法）
fn calculate_retry_interval() -> Duration {
    let retry_count = RETRY_COUNT.load(Ordering::SeqCst);
    let last_success_time = LAST_SUCCESS.load(Ordering::SeqCst);
    
    // 上次成功连接的时间（秒）
    let last_success_secs = if last_success_time > 0 {
        let now = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs() as u32;
        (now - last_success_time) as u64
    } else {
        u64::MAX // 尚未成功连接过
    };
    
    // 如果最近有成功连接，并且在稳定期内，使用最小重试间隔
    if last_success_secs < STABLE_PERIOD_MS / 1000 {
        trace!(
            last_success_secs,
            "In stable period, using minimum retry interval"
        );
        return Duration::from_millis(MIN_RETRY_INTERVAL_MS);
    }
    
    // 计算指数退避时间，但设置上限
    let base_ms = if retry_count > 0 {
        let exp = retry_count.saturating_sub(1).min(16); // 防止溢出
        let interval = MIN_RETRY_INTERVAL_MS * 2u64.pow(exp);
        trace!(
            retry_count,
            exponent = exp,
            calculated_ms = interval,
            "Calculated exponential backoff"
        );
        interval
    } else {
        trace!("No retries yet, using minimum interval");
        MIN_RETRY_INTERVAL_MS
    };
    
    let final_interval = base_ms.min(MAX_RETRY_INTERVAL_MS);
    trace!(
        base_interval_ms = base_ms,
        max_interval_ms = MAX_RETRY_INTERVAL_MS,
        final_interval_ms = final_interval,
        "Determined retry interval"
    );
    
    Duration::from_millis(final_interval)
}

// 连接成功后更新状态
fn update_connection_success() {
    // 记录成功时间
    let now = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as u32;
    
    let prev_success = LAST_SUCCESS.swap(now, Ordering::SeqCst);
    let prev_count = RETRY_COUNT.swap(0, Ordering::SeqCst);
    
    if prev_success > 0 {
        let elapsed = now - prev_success;
        debug!(
            previous_success_time = prev_success,
            current_time = now,
            seconds_since_last_success = elapsed,
            reset_retry_count = prev_count,
            "Updated connection success state"
        );
    } else {
        debug!(
            current_time = now,
            reset_retry_count = prev_count,
            "First successful connection"
        );
    }
    
    info!("Connection to OAP server successfully established");
}

// 监控连接健康状态
async fn connection_health_monitor() {
    let check_interval = Duration::from_secs(RETRY_COUNT_DECAY_PERIOD_MS / 1000);
    trace!(
        check_interval_secs = check_interval.as_secs(),
        "Starting connection health monitor"
    );
    
    loop {
        trace!("Health monitor sleeping for {:?}", check_interval);
        sleep(check_interval).await;
        trace!("Health monitor woke up, checking state");
        
        // 逐渐衰减重试计数，这样就算有临时故障，一段时间后重试间隔也会恢复
        let current_count = RETRY_COUNT.load(Ordering::SeqCst);
        if current_count > 0 {
            let new_count = current_count - 1;
            RETRY_COUNT.store(new_count, Ordering::SeqCst);
            
            // 记录当前的退避时间
            let current_interval = calculate_retry_interval();
            debug!(
                old_retry_count = current_count,
                new_retry_count = new_count,
                interval_ms = current_interval.as_millis(),
                "Decreased retry backoff count"
            );
        } else {
            trace!("No active retry count, nothing to decay");
        }
    }
}

#[tracing::instrument(skip_all)]
async fn connect(endpoint: Endpoint) -> Channel {
    debug!("Beginning connection attempt to OAP server");
    let channel = loop {
        trace!("Attempting to connect to endpoint: {:?}", endpoint);
        match endpoint.connect().await {
            Ok(channel) => {
                debug!("Connection successful");
                update_connection_success();
                break channel;
            }
            Err(err) => {
                // 增加重试计数
                let current_count = RETRY_COUNT.fetch_add(1, Ordering::SeqCst);
                let retry_interval = calculate_retry_interval();
                
                error!(
                    error.kind = ?err.to_string(),
                    error.details = ?err,
                    retry_count = current_count + 1,
                    retry_interval_ms = retry_interval.as_millis(),
                    "Connection to OAP server failed, will retry after {:?}", 
                    retry_interval
                );
                
                trace!("Sleeping before retry attempt");
                sleep(retry_interval).await;
                debug!("Retry timeout complete, attempting to reconnect");
            }
        }
    };

    let uri = &*endpoint.uri().to_string();
    info!(uri, "Skywalking server connected");
    
    // 打印channel的一些基本信息
    trace!(
        "Channel established: {:?}",
        channel
    );

    channel
}
